package tpc.task.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.ToolRunner;

import unisb.cs.core.binary.converter.BinaryConverter;
import unisb.cs.core.index.build.IndexBuilder;
import unisb.cs.core.join.cogroup.build.CogroupBuilder;
import unisb.cs.core.join.cogroup.index.CogroupIndexer;
import unisb.cs.core.join.cogroup.rearrange.CogroupRearranger;

public class JobUtils {

	public static void runBinaryConverter(String tableName, 
										  String inputPath, 
										  String outputPath,
										  String maxRecordLength, 
										  String numberOfReducer) throws Exception{
		
		long t1 = System.currentTimeMillis();
		
		String [] args = {tableName, inputPath, outputPath, maxRecordLength, numberOfReducer};
		Configuration jobConf = new Configuration();
		ToolRunner.run(jobConf, new BinaryConverter(), args);		
		
		long t2 = System.currentTimeMillis();
		System.out.println((float) (t2 - t1) / 1000);
	}
	
	public static void runIndexBuilder(String tableName, 
									   String inputPath, 
									   String outputPath,
									   String indexAttributeIndex, 
									   String indexAttributeType, 
									   String numberOfReducer) throws Exception{
		long t1 = System.currentTimeMillis();
		
		String [] args = {tableName, inputPath, outputPath, indexAttributeIndex, indexAttributeType, numberOfReducer};
		ToolRunner.run(new Configuration(), new IndexBuilder(), args);
		
		long t2 = System.currentTimeMillis();
		System.out.println((float) (t2 - t1) / 1000);
		
	}
	
	public static void runCogroupBuilder(String leftTableName, 
										 String leftTablePath, 
										 String rightTableName, 
										 String rightTablePath, 
										 String leftCogroupIndex,
										 String rightCogroupIndex,
										 String cogroupType,
										 String outputPath,
										 String numberOfReducer) throws Exception{
		long t1 = System.currentTimeMillis();
		
		String [] args = {leftTableName, leftTablePath, rightTableName, rightTablePath, 
				leftCogroupIndex, rightCogroupIndex, cogroupType, outputPath, numberOfReducer};	
		ToolRunner.run(new Configuration(), new CogroupBuilder(), args);
		
		long t2 = System.currentTimeMillis();
		System.out.println((float) (t2 - t1) / 1000);
		
	}
	
	public static void runRearrange(String leftTableName, 
			 						String rightTableName, 
			 						String inputPath, 
			 						String cogroupType,
			 						String outputPath,
			 						String numberOfReducer) throws Exception{
		long t1 = System.currentTimeMillis();		
		
		String [] args = {leftTableName, rightTableName, inputPath, cogroupType, outputPath, numberOfReducer};			
		ToolRunner.run(new Configuration(), new CogroupRearranger(), args);
		
		long t2 = System.currentTimeMillis();
		System.out.println((float) (t2 - t1) / 1000);
		
	}
	
	public static void runCogroupIndexer(String inputPath, 
										 String outputPath,
										 String tableName,
										 String leftRightSign,  //L or R
										 String indexAttributeIndex, 
							 			 String indexAttributeType, 
										 String numberOfReducer) throws Exception{
		long t1 = System.currentTimeMillis();
		
		String [] args = {inputPath, outputPath, tableName, leftRightSign, indexAttributeIndex, indexAttributeType, numberOfReducer};			
		ToolRunner.run(new Configuration(), new CogroupIndexer(), args);
		
		long t2 = System.currentTimeMillis();
		System.out.println((float) (t2 - t1) / 1000);
		
	}
	
	public static void mergeFiles(String path, String fileName){
		
	}
	
	
	//��Ҫ׼��
	//1��ִ��join��mapreduce���룬�̳���JobBase��������Ӧ���ṩһ���������ܹ�����һ������Ķ����������ʹ��ToolRunner.run
	//2����׼����InputFormat��RecordReader�������������������
	public static void runJob(JobBase job, String [] args) throws Exception{

		long t1 = System.currentTimeMillis();
		
		ToolRunner.run(new Configuration(), job, args);
		
		long t2 = System.currentTimeMillis();
		System.out.println((float) (t2 - t1) / 1000);
	}

}

